package belf.migrate.api.table.schema;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.common.Preconditions;
import belf.migrate.api.constants.SQLConstants;
import belf.migrate.api.exception.*;
import belf.migrate.api.model.*;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.util.CatalogUtils;
import belf.migrate.api.util.JdbcUrlUtil;
import belf.migrate.api.util.TableNameComparator;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * JDBC可访问的Catalog基类，定义了大部分的数据库访问操作
 */
@Slf4j
public abstract class AbstractJdbcCatalog implements Catalog {
    private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcCatalog.class);

    protected static final Set<String> SYS_DATABASES = new HashSet<>();

//    protected static final Set<String> SYS_SCHEMAS = new HashSet<>();

    /**
     * 表示数据库类型。定义参考：{@link DatabaseIdentifier}
     */
    protected final String catalogName;
    protected final String defaultDatabase;
    protected final String username;
    protected final String pwd;
    protected final String baseUrl;
    protected final String suffix;
    protected final String defaultUrl;
    protected final boolean caseSensitive;

    protected final Optional<String> defaultSchema;

    protected final Map<String, Connection> connectionMap;

    public AbstractJdbcCatalog(
            String catalogName,
            String username,
            String pwd,
            JdbcUrlUtil.UrlInfo urlInfo,
            String defaultSchema,
            boolean caseSensitive) {

        this.catalogName = catalogName;
        this.defaultDatabase = urlInfo.getDefaultDatabase().orElse(null);
        this.username = username;
        this.pwd = pwd;
        this.baseUrl = urlInfo.getUrlWithoutDatabase();
        this.defaultUrl = urlInfo.getOrigin();
        this.suffix = urlInfo.getSuffix();
        this.defaultSchema = Optional.ofNullable(defaultSchema);
        this.connectionMap = new ConcurrentHashMap<>();
        this.caseSensitive = caseSensitive;
    }

    @Override
    public String name() {
        return catalogName;
    }

    @Override
    public boolean caseSensitive() {
        return caseSensitive;
    }

    @Override
    public String getDefaultDatabase() {
        return defaultDatabase;
    }

    public Connection getConnection(String url) {
        if (connectionMap.containsKey(url)) {
            Connection connection = connectionMap.get(url);
            try {
                if (connection != null && !(connection.isClosed())) {
                    return connectionMap.get(url);
                }
            } catch (SQLException e) {
            }
        }

        try {
            Connection connection = DriverManager.getConnection(url, username, pwd);
            connectionMap.put(url, connection);
            return connection;
        } catch (Exception e) {
            log.error("Failed connecting to {} via JDBC. Exception:{}", url, e);
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void open() throws CatalogException {
        getConnection(defaultUrl);
        LOG.info("Catalog {} established connection to {}", catalogName, defaultUrl);
    }

    @Override
    public void close() throws CatalogException {
        for (Map.Entry<String, Connection> entry : connectionMap.entrySet()) {
            try {
                entry.getValue().close();
            } catch (SQLException e) {
                throw new CatalogException(
                        String.format("Failed to close %s via JDBC.", entry.getKey()), e);
            }
        }
        connectionMap.clear();
        LOG.info("Catalog {} closing", catalogName);
    }

    protected String getSelectColumnsSql(TablePath tablePath) {
        return "SELECT * FROM " + getFullTableName(tablePath);
    }

    protected String getSelectDataSql(TablePath tablePath, String pk, List<String> pkList) {
        return String.format("SELECT * FROM %s where %s in ('%s') order by %s", getFullTableName(tablePath), pk, String.join("','", pkList), pk);
    }

    protected String getDataSqlNoPk(TablePath tablePath, int limit, int offset) {
        throw new UnsupportedOperationException();
    }

    protected String getRandomPksSql(TablePath tablePath, String pk, int limit, int offset) {
        return String.format("select %s from %s limit %s,%s", pk, getFullTableName(tablePath), offset, limit);
    }

    protected ColumnMetaData buildColumn(ResultSet resultSet) throws SQLException {
        throw new UnsupportedOperationException();
    }

    protected TableIdentifier getTableIdentifier(TablePath tablePath) {
        return TableIdentifier.of(
                catalogName,
                tablePath.getDatabaseName(),
                tablePath.getSchemaName(),
                tablePath.getTableName(),
                tablePath.getQualifiersModel());
    }

    protected String getUrlFromDatabaseName(String databaseName) {
        if (defaultUrl.contains(databaseName)) {
            return defaultUrl;
        }

        String url = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";


        if (null == databaseName) {
            return url + suffix;
        }

        return url + databaseName + suffix;
    }

    @Override
    public List<IndexKey> getIndexKeys(TablePath tablePath) throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl;
        if (StringUtils.isNotBlank(tablePath.getDatabaseName())) {
            dbUrl = getUrlFromDatabaseName(tablePath.getDatabaseName());
        } else {
            dbUrl = getUrlFromDatabaseName(defaultDatabase);
        }
        Connection conn = getConnection(dbUrl);
        try {
            DatabaseMetaData metaData = conn.getMetaData();
            return CatalogUtils.getIndexKeys(metaData, tablePath);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    public CatalogTable getTable(TablePath tablePath)
            throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl;
        if (StringUtils.isNotBlank(tablePath.getDatabaseName())) {
            dbUrl = getUrlFromDatabaseName(tablePath.getDatabaseName());
        } else {
            dbUrl = getUrlFromDatabaseName(defaultDatabase);
        }

        Connection conn = getConnection(defaultUrl);

        try {
            DatabaseMetaData metaData = conn.getMetaData();
            // 查询主键
            PrimaryKey primaryKey = CatalogUtils.getPrimaryKey(metaData, tablePath);
            // 查询索引信息
            List<IndexKey> indexKeys = getIndexKeys(tablePath);
            // 查询外键
            List<ForeignKey> foreignKeys = CatalogUtils.getForeignKeys(metaData, tablePath);
            // 查询唯一不重复字段
            List<UniqueKey> uniqueKeys = getUniqueKeys(tablePath);
            // 查询字段额外的约束条件
            List<CheckConstraint> checkConstraints = getCheckConstraints(tablePath);

            // 查询字段信息
            String selectColumnsSQL = getSelectColumnsSql(tablePath);
            log.debug("Query columns: {}", selectColumnsSQL);
            try (PreparedStatement ps = conn.prepareStatement(selectColumnsSQL);
                 ResultSet resultSet = ps.executeQuery()) {

                ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
                // 查询表的字段基本信息，这个是基于JDBC API查询，查询的是各种关系库最小子集
                List<ColumnMetaData> columns = CatalogUtils.getColumns(resultSetMetaData, tablePath);
                // 查询表的字段的补充信息，这个跟各个数据库相关，每个源端Catalog自己实现
                fillingColumnExtra(tablePath, columns, conn);

                TableSchema tableSchema = TableSchema
                        .builder()
                        .columns(columns)
                        .primaryKey(primaryKey)
                        .indexKeys(indexKeys)
                        .foreignKeys(foreignKeys)
                        .uniqueKeys(uniqueKeys)
                        .checkConstraints(checkConstraints)
                        .build();

                TableIdentifier tableIdentifier = getTableIdentifier(tablePath);
                return CatalogTable.of(
                        tableIdentifier,
                        tableSchema,
                        buildConnectorOptions(tablePath),
                        "",
                        catalogName);
            }
        } catch (Exception e) {
            log.error("Failed getting table {}, message: {}", tablePath.getFullName(), e.getMessage());
            e.printStackTrace();
            throw new CatalogException("Failed getting table " + tablePath.getFullName(), e);
        }
    }

    /**
     * 源端的Catalog需要重载这个方法，补充完整字段默认值、Extra信息、字段备注信息
     * @param columns
     * @param connection
     */
    protected void fillingColumnExtra(TablePath tablePath, List<ColumnMetaData> columns, Connection connection) {
    }

    public List<List<ColumnData>> getRandomDataFromTable(TablePath tablePath, String pkName, List<String> pkList)
            throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        Connection conn = getConnection(defaultUrl);

        try {
            String selectDataSql = getSelectDataSql(tablePath, pkName, pkList);
            log.debug("Query data: {}", selectDataSql);
            try (PreparedStatement ps = conn.prepareStatement(selectDataSql);
                 ResultSet resultSet = ps.executeQuery()) {

                List<ColumnMetaData> columns = getAllColumnsBySql(tablePath);
                ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
                List<ColumnMetaData> jdbcColumns = CatalogUtils.getColumns(resultSetMetaData, tablePath);
                Map<String, Integer> jdbcColMap = jdbcColumns.stream().collect(Collectors.toMap(ColumnMetaData::getColumnName, ColumnMetaData::getColumnType));
                List<List<ColumnData>> rows = new ArrayList<>();
                while (resultSet.next()) {
                    List<ColumnData> row = new ArrayList<>();
                    for (int i = 1; i <= columns.size(); i++) {
                        ColumnMetaData columnMetaData = columns.get(i - 1);
                        ColumnData columnData = new ColumnData();
                        columnData.setColumnName(columnMetaData.getColumnName());
                        String columnType = columnMetaData.getColumnTypeName();
                        columnData.setColumnTypeName(columnType);
                        columnData.setColumnType(jdbcColMap.get(columnData.getColumnName()));
                        columnMetaData.setColumnType(jdbcColMap.get(columnData.getColumnName()));
                        try {
                            if (bigDataColumns().contains(columnType.toUpperCase(Locale.ROOT))) {
                                byte[] bytes = resultSet.getBytes(i);
                                columnData.setColumnValue(bytes == null ? 0 :bytes.length);
                            } else {
                                columnData.setColumnValue(getColumnValueByType(columnMetaData, resultSet, i));
                            }
                        } catch (SQLException e) {
                            log.error("字段类型暂不支持！", e);
                        }
                        row.add(columnData);
                    }
                    rows.add(row);
                }
                return rows;
            }
        } catch (Exception e) {
            log.error("Failed getting table {}, message: {}", tablePath.getFullName(), e.getMessage());
            e.printStackTrace();
            throw new CatalogException("Failed getting table " + tablePath.getFullName(), e);
        }
    }

    public Object getColumnValueByType(ColumnMetaData columnMetaData, ResultSet resultSet, int index) throws SQLException {
        Object data = null;
        try {
            data = switch (columnMetaData.getColumnType()) {
                case Types.TIMESTAMP -> resultSet.getTimestamp(index);
                case Types.DATE -> resultSet.getDate(index);
                case Types.TIME -> resultSet.getTime(index);
                default -> resultSet.getObject(index);
            };
        } catch (Exception e) {
            try {
                data = resultSet.getString(index);
            } catch (SQLException ex) {
                log.error("字段类型暂不支持！", e);
            }
        }
        return data;
    }

    public List<List<ColumnData>> getDataWithNoPk(TablePath tablePath, int limit, int offset)
            throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        Connection conn = getConnection(defaultUrl);

        try {
            String selectDataSql = getDataSqlNoPk(tablePath, limit, offset);
            log.debug("Query data: {}", selectDataSql);
            try (PreparedStatement ps = conn.prepareStatement(selectDataSql);
                 ResultSet resultSet = ps.executeQuery()) {

                List<ColumnMetaData> columns = getAllColumnsBySql(tablePath);
                List<List<ColumnData>> rows = new ArrayList<>();
                while (resultSet.next()) {
                    List<ColumnData> row = new ArrayList<>();
                    for (int i = 1; i <= columns.size(); i++) {
                        ColumnData columnData = new ColumnData();
                        columnData.setColumnName(columns.get(i - 1).getColumnName());
                        String columnType = columns.get(i - 1).getColumnTypeName();
                        columnData.setColumnTypeName(columnType);
                        try {
                            if (bigDataColumns().contains(columnType.toUpperCase(Locale.ROOT))) {
                                byte[] bytes = resultSet.getBytes(i);
                                columnData.setColumnValue(bytes == null ? 0 :bytes.length);
                            } else {
                                columnData.setColumnValue(getColumnValueByType(columns.get(i - 1), resultSet, i));
                            }
                        } catch (SQLException e) {
                            log.error("字段类型暂不支持！", e);
                        }
                        row.add(columnData);
                    }
                    rows.add(row);
                }
                return rows;
            }
        } catch (Exception e) {
            log.error("Failed getting table {}, message: {}", tablePath.getFullName(), e.getMessage());
            e.printStackTrace();
            throw new CatalogException("Failed getting table " + tablePath.getFullName(), e);
        }
    }

    public List<String> getRandomPks(TablePath tablePath, String pkName, int limit, int offset)
            throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        try {
            return queryString(
                    defaultUrl,
                    getRandomPksSql(tablePath, pkName, limit, offset),
                    rs -> rs.getString(1));
        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(
                    String.format("Failed get primary key data in catalog %s", this.catalogName), e);
        }
    }

    @SuppressWarnings("MagicNumber")
    protected Map<String, String> buildConnectorOptions(TablePath tablePath) {
        Map<String, String> options = new HashMap<>(8);
        options.put("connector", "jdbc");
        options.put("url", defaultUrl);
        options.put("table-name", getOptionTableName(tablePath));
        options.put("username", username);
        options.put("password", pwd);
        return options;
    }

    protected String getOptionTableName(TablePath tablePath) {
        return tablePath.getFullName();
    }

    public String getTableName(TablePath tablePath) {
        return tablePath.getSchemaAndTableName();
    }

    protected String getTableNameQuoted(String tableName) {
        return getQuote() + tableName + getQuote();

    }

    protected String getColumnNameQuoted(String columnName){
        return getQuote() + columnName + getQuote();
    }

    @Override
    public boolean tableExists(TablePath tablePath) throws CatalogException {
        try {
            List<String> tables = listTables(tablePath.getRightDatabaseName(), tablePath.getSchemaName());

            return TableNameComparator.containsTable(tables, tablePath.getTableName());
        } catch (DatabaseNotExistException e) {
            return false;
        }
    }

    @Override
    public boolean functionExists(String databaseName, String schemaName, String functionName) throws CatalogException {
        List<String> functionNames = listFunctionNames(databaseName, schemaName);
        if (functionNames != null) {
            for (String func : functionNames) {
                if (func.equalsIgnoreCase(functionName)) {
                    return true;
                }
            }
        }
        return false;
    }

    @Override
    public boolean viewExists(String databaseName, String schemaName, String viewName) throws CatalogException {
        String name = StringUtils.isNotEmpty(databaseName) ? databaseName : schemaName;
        List<String> views;
        try {
            views = listViewNames(databaseName, schemaName);
            if (views != null) {
                for (String view : views) {
                    if (view.equalsIgnoreCase(viewName)) {
                        return true;
                    }
                }
            }
        } catch (CatalogException e) {
            log.error("Failed to list views in database %s", name);
            return false;
        }
        return false;
    }

    @Override
    public boolean triggerExists(String databaseName, String schemaName, String triggerName) throws CatalogException {
        String name = StringUtils.isNotEmpty(databaseName) ? databaseName : schemaName;
        List<String> triggers;
        try {
            triggers = listTriggerNames(databaseName, schemaName);
            if (triggers != null) {
                for (String trigger : triggers) {
                    if (trigger.equalsIgnoreCase(triggerName)) {
                        return true;
                    }
                }
            }
        } catch (CatalogException e) {
            log.error("Failed to list triggers in database %s", name);
            return false;
        }
        return false;
    }

    @Override
    public boolean procedureExists(String databaseName, String schemaName, String procedureName) throws CatalogException {
        String name = StringUtils.isNotEmpty(databaseName) ? databaseName : schemaName;
        List<String> procedures;
        try {
            procedures = listProcedureNames(databaseName, schemaName);
            if (procedures != null) {
                for (String procedure : procedures) {
                    if (procedure.equalsIgnoreCase(procedureName)) {
                        return true;
                    }
                }
            }
        } catch (CatalogException e) {
            log.error("Failed to list procedures in database %s", name);
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR,
                    String.format("Failed listing procedure in catalog %s", this.catalogName));
        }
        return false;
    }

    @Override
    public boolean typeExists(String databaseName, String schemaName, String typeName) throws CatalogException {
        List<String> types;
        types = listTypeNames(databaseName, schemaName);
        if (types != null) {
            for (String procedure : types) {
                if (procedure.equalsIgnoreCase(typeName)) {
                    return true;
                }
            }
        }
        return false;
    }

    @Override
    public boolean sequenceExists(String databaseName, String schemaName, String sequenceName) throws CatalogException {
        List<String> sequences;
        sequences = listSeqNames(databaseName, schemaName);
        if (sequences != null) {
            for (String procedure : sequences) {
                if (procedure.equalsIgnoreCase(sequenceName)) {
                    return true;
                }
            }
        }
        return false;
    }

    protected void closeDatabaseConnection(String databaseName) throws CatalogException {
        String dbUrl = getUrlFromDatabaseName(databaseName);
        try {
            Connection connection = connectionMap.remove(dbUrl);
            if (connection != null) {
                connection.close();
            }
        } catch (SQLException e) {
            throw new CatalogException(String.format("Failed to close %s via JDBC.", dbUrl), e);
        }
    }

    @Override
    public void createTable(TablePath tablePath, CatalogTable table, boolean ignoreIfExists, QualifiersModel qualifiersModel, TaskConf taskConf)
            throws TableAlreadyExistException, CatalogException, CreateTableException, DatabaseNotExistException {
        Preconditions.checkNotNull(tablePath, "Table path cannot be null");

        String databaseName = tablePath.getRightDatabaseName();
        if (null == databaseName) databaseName = tablePath.getSchemaName();
        if (!databaseExists(databaseName)) {
            throw new DatabaseNotExistException(catalogName, databaseName);
        }
        if (defaultSchema.isPresent()) {
            tablePath =
                    new TablePath(
                            tablePath.getDatabaseName(),
                            defaultSchema.get(),
                            tablePath.getTableName(),
                            qualifiersModel);
        }

        if (tableExists(tablePath)) {
            if (ignoreIfExists) {
                return;
            }
            throw new TableAlreadyExistException(catalogName, tablePath);
        }

        createTableInternal(tablePath, table, taskConf);
    }

    public String getCreateTableSql(TablePath tablePath, CatalogTable table, TaskConf taskConf) throws DataTypeMappingNotExistException {
        throw new UnsupportedOperationException();
    }

    protected void createTableInternal(TablePath tablePath, CatalogTable table, TaskConf taskConf)
            throws CreateTableException {
        String dbUrl = defaultUrl;
        String createTableSql = null;
        try {
            createTableSql = getCreateTableSql(tablePath, table, taskConf);
        } catch (DataTypeMappingNotExistException e) {
            throw new CreateTableException(table.getCatalogName(), tablePath, e.getCause(), createTableSql);
        }
        try {
            executeInternal(dbUrl, createTableSql);
            log.info("Created table {}, sql:\n{}", table.getTableId().getTableName(), createTableSql);
        } catch (CatalogException e) {
            log.info("Create table {} error, sql:\n{}", table.getTableId().getTableName(), createTableSql);
            throw new CreateTableException(table.getCatalogName(), tablePath, e.getCause(), createTableSql);
        }
    }

    public void createIndex(TablePath tablePath, CatalogTable table, boolean ignoreIfNotExists, TaskConf taskConf)
            throws IndexAlreadyExistException, CreateIndexException {
        String dbUrl = defaultUrl;

        // TODO：先检查索引是否存在了

        List<String> indexSqls = getCreateIndexSqls(tablePath, table, taskConf);
        for (String indexSql : indexSqls) {
            try {
                executeInternal(dbUrl, indexSql);
                log.info("Create Index success On {}, sql={}", table.getTableId().getTableName(), indexSql);
            } catch (Exception e) {
                log.error("Create Index error On {}, sql={}, message: {}", table.getTableId().getTableName(), indexSql, e.getMessage());
                e.printStackTrace();
                throw new CreateIndexException(table.getCatalogName(), tablePath, e, indexSql);
            }

        }
    }

    public boolean checkIndexExists(String url, String tableName, String indexName) throws SQLException {
        Connection connection = getConnection(url);
        DatabaseMetaData metaData = connection.getMetaData();
        try (ResultSet indexes = metaData.getIndexInfo(null, null, tableName, false, false)) {
            while (indexes.next()) {
                String currentIndexName = indexes.getString("INDEX_NAME");
                if (indexName.equals(currentIndexName)) {
                    return true;
                }
            }
        }
        return false;
    }

    public void createForeignKey(TablePath tablePath, CatalogTable table, boolean ignoreIfNotExists, TaskConf taskConf) throws CatalogException {
        String dbUrl = defaultUrl;
        List<String> indexSqls = getCreateForeignSqls(tablePath, table, taskConf);
        for (String indexSql : indexSqls) {
            executeInternal(dbUrl, indexSql);
            log.info("Create ForeigKey On {}, sql={}", table.getTableId().getTableName(), indexSql);
        }
    }

    public List<String> getCreateIndexSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        throw new UnsupportedOperationException();
    }

    public List<String> getCreateForeignSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        throw new UnsupportedOperationException();
    }


    @Override
    public String showCreateTable(TablePath tablePath) throws TableNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateView(TablePath tablePath) throws ViewNotExistException {
        throw new UnsupportedOperationException();
    }

    public String showCreateProcedure(TablePath tablePath) throws ProcedureNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateTrigger(TablePath tablePath) throws TriggerNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateType(TablePath tablePath) throws TypeNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateSeq(TablePath tablePath) throws SequenceNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateFunc(TablePath tablePath) throws FunctionNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateIndex(TablePath tablePath) throws IndexNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public String showCreateSynonym(TablePath tablePath) throws SequenceNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void dropTable(TablePath tablePath, boolean ignoreIfNotExists)
            throws TableNotExistException, CatalogException {
        Preconditions.checkNotNull(tablePath, "Table path cannot be null");

        if (!tableExists(tablePath)) {
            if (ignoreIfNotExists) {
                return;
            }
            throw new TableNotExistException(catalogName, tablePath);
        }

        dropTableInternal(tablePath);
    }

    public String getDropTableSql(TablePath tablePath) {
        return "DROP TABLE " + tablePath.getFullName();
    }

    public String getDropViewSql(TablePath viewPath) {
        return String.format("DROP VIEW IF EXISTS %s",
                viewPath.getFullNameWithQuoted());
    }

    public String getDropTriggerSql(TablePath triggerPath) {
        return String.format("DROP TRIGGER IF EXISTS %s",
                triggerPath.getFullName());
    }

    public String getDropTypeSql(TablePath typePath) {
        return String.format("DROP TYPE %s", typePath.getFullName());
    }

    public String getDropSequenceSql(TablePath seqPath) {
        return String.format("DROP SEQUENCE %s", seqPath.getFullName());
    }

    public String getDropFunctionSql(TablePath functionPath) {
        return String.format("DROP FUNCTION %s", functionPath.getFullName());
    }

    public String getDropSynonymSql(TablePath synonymPath) {
        // FIXME: 注意补齐对公有同义词的删除处理：是否需要加关键词“PUBLIC”
        return String.format("DROP SYNONYM %s", synonymPath.getFullName());
    }

    public String getDropProcedureSql(TablePath procedurePath) {
        return String.format("DROP PROCEDURE %s", procedurePath.getFullName());
    }

    protected String getDropTypeSql(String fullTypeName) {
        return String.format("DROP TYPE %s CASCADE CONSTRAINTS", fullTypeName);
    }

    protected void dropTableInternal(TablePath tablePath) throws CatalogException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        // Will there exist concurrent drop for one table?
        executeInternal(dbUrl, getDropTableSql(tablePath));
    }


    // If sql is DDL, the execute() method always returns false, so the return value
    // should not be used to determine whether changes were made in database.
    protected boolean executeInternal(String url, String sql) throws CatalogException {
//        LOG.info("Execute sql : {}", sql);
        try (PreparedStatement ps = getConnection(url).prepareStatement(sql)) {
            ps.execute();
            return true;
        } catch (Exception e) {
            log.error("Execute sql error: \nsql={}  \nerror: {}", sql, e.getMessage());
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, String.format("Failed executeSql error: \n%s", sql), e);
        }
    }


    protected String getTruncateTableSql(TablePath tablePath) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查看指定表下是否有数据存在。<br/> 对于mysql、dameng，是支持"select * from table limit 1"的。其他数据库请自己定义。<br/>
     * 注意：对于无主键表，不能用select * count(*) 去判断是否存在数据，性能会非常差
     *
     * @param tablePath
     * @return
     */
    public String getExistDataSql(TablePath tablePath) {
        if (caseSensitive()) {
            return String.format(
                    "SELECT * FROM %s.%s LIMIT 1;",
                    tablePath.getRightDatabaseName().toUpperCase(), tablePath.getTableName().toUpperCase());
        } else {
            return String.format(
                    "SELECT * FROM %s.%s LIMIT 1;",
                    tablePath.getRightDatabaseName(), tablePath.getTableName());
        }
    }

    /**
     * 执行SELECT这样的DML语句
     *
     * @param tablePath
     * @param sql
     * @throws CatalogException
     */
    public void executeSql(TablePath tablePath, String sql) throws CatalogException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            // Will there exist concurrent drop for one table?
            ps.execute();
        } catch (SQLException e) {
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, String.format("Failed executeSql error:\n %s", sql), e);
        }
    }

    /**
     * 一个通用的方法，查询特定数据库对象的名称列表，如：序列、类型、触发器、函数、视图等
     * @param sql 查询对象名称的sql
     * @param typeName 按指定名称读返回的结果（从{@link ResultSet}中按这个名字读取）
     * @return
     * @throws CatalogException
     */
    public List<String> listNames(String sql, String typeName) throws CatalogException {
        List<String> names = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            // Will there exist concurrent drop for one table?
            ps.executeQuery();
            try (ResultSet rs = ps.getResultSet()) {
                while (rs.next()) {
                    String name = rs.getString(typeName);
                    names.add(name);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, String.format("Failed executeSql error: %s", sql), e);
        }
        return names;
    }

    /**
     * 查询创建数据库对象的DDL
     * @param sql 查询语句
     * @param ddlName 查询结果中ddl语句的字段名称
     * @return
     * @throws CatalogException
     */
    public String getDDL(String sql, String ddlName) throws CatalogException {
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            // Will there exist concurrent drop for one table?
            ps.executeQuery();
            try (ResultSet rs = ps.getResultSet()) {
                if (rs.next()) {
                    return rs.getString(ddlName);
                }
            }
        } catch (SQLException e) {
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, String.format("Failed executeSql error: \n%s", sql), e);
        }
        return null;
    }

    /**
     * 执行SQL语句 ，可获取执行结果
     * @param tablePath
     * @param sql
     * @return
     * @throws SQLException
     */
    public List<JdbcExecuteResult> execute(TablePath tablePath, String sql) throws SQLException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        List<JdbcExecuteResult> resultList = null;
        try (Statement statement = connection.createStatement()) {
            boolean hasResultSet = statement.execute(sql);
            if (hasResultSet) {
                resultList = new ArrayList<>();
            }
            while (hasResultSet) {
                JdbcExecuteResult jdbcExecuteResult = new JdbcExecuteResult();
                try (ResultSet resultSet = statement.getResultSet()) {
                    List<ColumnMetaData> columns = CatalogUtils.getColumns(resultSet.getMetaData(), tablePath);
//                    List<Map<String, String>> rows = new ArrayList<>();
                    List<Map<String, Object>> rows = new ArrayList<>();
                    jdbcExecuteResult.setColumns(columns);
                    int columnSize = columns.size();
                    while (resultSet.next()) {
//                        Map<String, String> resultMap = new LinkedHashMap<>();
                        Map<String, Object> resultMap = new LinkedHashMap<>();
                        for (int i = 1; i <= columnSize; i++) {
//                            resultMap.put(columns.get(i - 1).getColumnName(), resultSet.getString(i));
                            resultMap.put(columns.get(i - 1).getColumnName(), resultSet.getObject(i));
                        }
                        rows.add(resultMap);
                    }
                    jdbcExecuteResult.setRows(rows);
                }
                resultList.add(jdbcExecuteResult);
                hasResultSet = statement.getMoreResults();
            }
        }
        return resultList;
    }

    /**
     * 执行DDL语句。错误情况通过异常返回
     * @param tablePath
     * @param ddl
     * @throws DDLException
     */
    @Override
    public void executeDdl(TablePath tablePath, String ddl) throws DDLException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        try {
            Statement stmt = connection.createStatement();
            stmt.executeUpdate(ddl);
            stmt.close();
        } catch (SQLException e) {
            throw new DDLException("执行DDL失败", e, ddl);
        }
    }

    public boolean isExistsData(TablePath tablePath) throws CatalogException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        String sql = getExistDataSql(tablePath);
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ResultSet resultSet = ps.executeQuery();
            return resultSet.next();
        } catch (SQLException e) {
            throw new CatalogException(String.format("Failed executeSql error %s", sql), e);
        }
    }

    @Override
    public String countDataForTable(TablePath tablePath) throws CatalogException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        String sql = getCountDataSql(tablePath);
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ResultSet resultSet = ps.executeQuery();
            resultSet.next();
            return resultSet.getString(1);
        } catch (SQLException e) {
            throw new CatalogException(String.format("Failed executeSql error %s", sql), e);
        }
    }

    @Override
    public boolean databaseExists(String databaseName) throws CatalogException {
        Preconditions.checkArgument(StringUtils.isNotBlank(databaseName));

        return TableNameComparator.containsDatabase(listDatabases(), databaseName);
    }

    @Override
    public List<String> listDatabases() throws CatalogException {
        try {
            return queryString(
                    defaultUrl,
                    getListDatabaseSQL(),
                    rs -> {
                        String s = rs.getString(1).trim();
                        return SYS_DATABASES.contains(s) ? null : s;
                    });
        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(
                    String.format("Failed listing database in catalog %s", this.catalogName), e);
        }
    }

    @Override
    public List<String> listCatalogs() throws CatalogException {
        try {
            return queryString(
                    defaultUrl,
                    getListCatalogSQL(),
                    rs -> {
                        String s = rs.getString(1);
                        return SYS_DATABASES.contains(s) ? null : s;
                    });
        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(
                    String.format("Failed listing catalog in database %s", this.catalogName), e);
        }
    }

    @Override
    public List<String> listSchemasByDatabase(String databaseName) throws CatalogException {
        throw new UnsupportedOperationException();
    }



    protected String getListDatabaseSQL() {
        throw new UnsupportedOperationException();
    }

    protected String getListCatalogSQL() {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName) throws CatalogException, DatabaseNotExistException {
        throw new UnsupportedOperationException();
    }

    @Override
    public TableInfoModel getTableInfoByName(TablePath tablePath) throws CatalogException {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<String> listViewNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabaseName = schemaName != null? schemaName : databaseName;
        String sql = getListViewsSQL(defaultDatabaseName);
        return listNames(sql, "VIEW_NAME");

    }

    @Override
    public List<ViewInfoModel> listViewsInfo(String databaseName, String schemaName) throws CatalogException {
        List<ViewInfoModel> viewInfoModels = new ArrayList<>();

        try (PreparedStatement ps = getConnection(defaultUrl).prepareStatement(getListViewsSQLInfo(databaseName, schemaName))) {

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                ViewInfoModel viewInfoModel = new ViewInfoModel();
                viewInfoModel.setViewName(rs.getString("VIEW_NAME"));
                viewInfoModels.add(viewInfoModel);
            }

        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR,
                    String.format("Failed listing view in catalog %s", this.catalogName));
        }
        return viewInfoModels;
    }

    @Override
    public List<String> listFunctionNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabase = schemaName != null? schemaName: databaseName;
        String sql = getListFunctionSQL(defaultDatabase);
        return listNames(sql, "FUNCTION_NAME");
    }

    @Override
    public List<String> listProcedureNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabase = schemaName != null? schemaName: databaseName;
        String sql = getListProcedureSQL(defaultDatabase);
        return listNames(sql, "PROCEDURE_NAME");
    }

    @Override
    public List<ProcedureInfoModel> listProceduresInfo(String databaseName, String schemaName) throws CatalogException {
        List<ProcedureInfoModel> procedureInfoModels = new ArrayList<>();

        String defaultDatabase = schemaName != null? schemaName: databaseName;
        try (PreparedStatement ps = getConnection(defaultUrl).prepareStatement(getListProcedureSQL(defaultDatabase))) {

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                ProcedureInfoModel procedureInfoModel = new ProcedureInfoModel();
                procedureInfoModel.setProcedureName(rs.getString("PROCEDURE_NAME"));
                procedureInfoModels.add(procedureInfoModel);
            }

        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR,
                    String.format("Failed listing procedure in catalog %s", this.catalogName));
        }
        return procedureInfoModels;
    }

    @Override
    public List<String> listSeqNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabase = schemaName != null? schemaName: databaseName;
        String sql = getListSequenceSQL(defaultDatabase);
        return listNames(sql, "SEQUENCE_NAME");
    }

    @Override
    public List<String> listSynonymNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabase = schemaName != null? schemaName: databaseName;
        String sql = getListSynonymSQL(defaultDatabase);
        return listNames(sql, "SYNONYM_NAME");
    }

    @Override
    public List<String> listTriggerNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabase = schemaName != null? schemaName: databaseName;
        String sql = getListTriggerSQL(defaultDatabase);
        return listNames(sql, "TRIGGER_NAME");
    }

    @Override
    public List<IndexInfo> listIndexInfos(String databaseName, String schemaName) throws CatalogException {
        List<IndexInfo> indexInfos = new ArrayList<>();
        String defaultDatabase = schemaName != null? schemaName: databaseName;
        try (PreparedStatement ps = getConnection(defaultUrl).prepareStatement(getListIndexSQL(defaultDatabase))) {

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                IndexInfo indexInfo = new IndexInfo();
                indexInfo.setIndexName(rs.getString("INDEX_NAME"));
                indexInfo.setTableName(rs.getString("TABLE_NAME"));
                indexInfo.setIndexType(rs.getString("INDEX_TYPE"));
                indexInfos.add(indexInfo);
            }

        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR,
                    String.format("Failed listing index in catalog %s", this.catalogName));
        }
        return indexInfos;
    }

    @Override
    public List<TriggerInfoModel> listTriggersInfo(String databaseName, String schemaName) throws CatalogException {
        List<TriggerInfoModel> triggerInfoModels = new ArrayList<>();

        String defaultDatabase = schemaName != null ? schemaName : databaseName;
        try (PreparedStatement ps = getConnection(defaultUrl).prepareStatement(getListTriggerSQL(defaultDatabase))) {

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                TriggerInfoModel triggerInfoModel = new TriggerInfoModel();
                triggerInfoModel.setTriggerName(rs.getString("TRIGGER_NAME"));
                triggerInfoModels.add(triggerInfoModel);
            }

        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR,
                    String.format("Failed listing trigger in catalog %s", this.catalogName));
        }
        return triggerInfoModels;
    }

    /**
     * 返回查询所有类型TYPE的SQL语句
     * @param databaseName
     * @param schemaName
     * @return
     * @throws CatalogException
     */
    @Override
    public List<String> listTypeNames(String databaseName, String schemaName) throws CatalogException {
        String defaultDatabase = schemaName != null ? schemaName : databaseName;
        String sql = getListTypeSQL(defaultDatabase);
        return listNames(sql, "TYPE_NAME");
    }

    @Override
    public List<TypeInfoModel> listTypesInfo(String databaseName, String schemaName) throws CatalogException {
        List<TypeInfoModel> typeInfoModels = new ArrayList<>();
        String defaultDatabase = schemaName != null ? schemaName : databaseName;
        try (PreparedStatement ps = getConnection(defaultUrl).prepareStatement(getListTypeSQL(defaultDatabase))) {

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                TypeInfoModel typeInfoModel = new TypeInfoModel();
                typeInfoModel.setTypeName(rs.getString("TYPE_NAME"));
                typeInfoModels.add(typeInfoModel);
            }

        } catch (UnsupportedOperationException e) {
            return null;
        } catch (Exception e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_TYPE_LIST_ERROR, "查询TYPE列表失败", e);
        }
        return typeInfoModels;


    }

    /**
     * 查询所有表名称的SQL语句
     * @param database
     * @param schema
     * @return
     */
    protected String getListTableSQL(String database, String schema) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有视图名称的SQL语句
     * @param defaultDatabase
     * @return
     */
    protected String getListViewsSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有视图名称的SQL语句
     * @param defaultDatabase
     * @param schemaName
     * @return
     */
    protected String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有存储过程名称的SQL语句
     * @param defaultDatabase
     * @return
     */
    protected String getListProcedureSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有函数名称的SQL语句
     * @param defaultDatabase
     * @return
     */
    protected String getListFunctionSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有表名称的SQL语句
     * @param databaseName
     * @param schemaName
     * @return
     */
    @Override
    public List<String> listTables(String databaseName, String schemaName)
            throws CatalogException, DatabaseNotExistException {
        String sql = getListTableSQL(databaseName, schemaName);
        return listNames(sql, "TABLE_NAME");
    }

    /**
     * 查询所有序列名称的SQL语句
     * @param defaultDatabase
     * @return
     */
    protected String getListSequenceSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有同义词名称的SQL语句
     * @param defaultDatabase
     * @return
     */
    protected String getListSynonymSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    /**
     * 查询所有触发器名称的SQL语句
     * @param defaultDatabase
     * @return
     */
    protected String getListTriggerSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    protected String getListIndexSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    protected String getListTypeSQL(String defaultDatabase) {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<String> specialColumns() {
        return Collections.emptyList();
    }

    @Override
    public List<String> bigDataColumns() {
        return Collections.emptyList();
    }

    @Override
    public List<ColumnMetaData> getAllColumnsBySql(TablePath tablePath) throws CatalogException, TableNotExistException {
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        String sql = getColumnsSql(tablePath);
        List<ColumnMetaData> result = new ArrayList<>();
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ResultSet resultSet = ps.executeQuery();
            while (resultSet.next()) {
                ColumnMetaData column = ColumnMetaData.builder()
                        .catalogName(resultSet.getString("catalog_name"))
                        .schemaName(resultSet.getString("schema_name"))
                        .tableName(resultSet.getString("table_name"))
                        .columnName(resultSet.getString("column_name"))
                        .columnTypeName(resultSet.getString("column_type"))
                        .columnDisplaySize(StringUtils.isEmpty(resultSet.getString("data_length")) ? 0 : Long.parseLong(resultSet.getString("data_length")))
                        .nullable(resultSet.getInt("nullable"))
                        .columnComment(resultSet.getString("column_comment"))
                        .isPrimaryKey(resultSet.getBoolean("is_primary_key"))
                        .build();
                result.add(column);
            }
            fillingColumnExtra(tablePath, result, connection);
            return result;
        } catch (SQLException e) {
            throw new CatalogException(String.format("Failed executeSql error %s", sql), e);
        }
    }

    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        throw new CatalogException(APIErrorCode.METHOD_NOT_IMPLEMENTED, "方法getUniqueKeys()还没有实现");
    }

    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        throw new CatalogException(APIErrorCode.METHOD_NOT_IMPLEMENTED, "方法getCheckConstraints()还没有实现");
    }

    /**
     * 查询表的字段数据类型（跟数据库相关）的SQL语句
     * @param tablePath
     * @return
     */
    protected String getColumnsSql(TablePath tablePath) {
        return null;
    }

    @FunctionalInterface
    public interface ResultSetConsumer<T> {
        T apply(ResultSet rs) throws SQLException;
    }

    @Override
    public List<String> findFunctionIncompatible(String sql, String sourceCatalog, String sinkCatalog) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean columnTypeEqual(String sourceType, String sinkType) {
        return sourceType.equalsIgnoreCase(sinkType);
    }

    @Override
    public boolean dataEqualByType(String type, Object sourceData, Object sinkData) {
        return sourceData.toString().equals(sinkData.toString());
    }

    /**
     * 设置当前数据库
     * @param connection
     * @param database
     * @throws SQLException
     */
    @Override
    public void setCurrentDatabase(Connection connection, String database) throws SQLException {
    }

    /**
     * 设置当前schema
     * @param connection
     * @param schema
     * @throws SQLException
     */
    @Override
    public void setCurrentSchema(Connection connection, String schema) throws SQLException {

    }

    protected List<String> queryString(String url, String sql, ResultSetConsumer<String> consumer)
            throws SQLException {
        try (PreparedStatement ps = getConnection(url).prepareStatement(sql)) {
            List<String> result = new ArrayList<>();
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                String value = consumer.apply(rs);
                if (value != null) {
                    result.add(value);
                }
            }
            return result;
        }
    }

    @Override
    public boolean isAuditLog() {
        return false;
    }

    @Override
    public List<CharacteristicInfo> getCharacteristicInfo(String sql) {
        return null;
    }

    /**
     * 去掉某些特定数据库特定sql结尾的分号
     */
    @Override
    public String trimSql(String sql) {
        // This is called only when use selects query (i.e. no automatic query detection)
        if (sql.isEmpty()) {
            // Do not trim delimiter
            return sql;
        }

        // Here we cur trailing query delimiter. most of DBs don't expect it in the end of query
        // However Oracle REQUIRES that block queries (e.g. DDL like CREATE PROCEDURE) must have trailing delimiter
        // So we check whether this query is a block query (by checking for all SQL dialect block delimiters)
        String trailingSpaces = "";
        {
            int trailingSpacesCount = 0;
            for (int i = sql.length() - 1; i >= 0; i--) {
                if (!Character.isWhitespace(sql.charAt(i))) {
                    break;
                }
                trailingSpacesCount++;
            }
            if (trailingSpacesCount > 0) {
                trailingSpaces = sql.substring(sql.length() - trailingSpacesCount);
                sql = sql.substring(0, sql.length() - trailingSpacesCount);
            }
        }
        String[] statementDelimiters;
        statementDelimiters = getScriptDelimiters();
        for (String statementDelimiter : statementDelimiters) {
            if (!sql.endsWith(statementDelimiter) && sql.length() > statementDelimiter.length()) {
                continue;
            }
            if (Character.isAlphabetic(statementDelimiter.charAt(0))) {
                // Delimiter is alphabetic (e.g. "GO") so it must be prefixed with whitespace
                char lastChar = sql.charAt(sql.length() - statementDelimiter.length() - 1);
                if (Character.isUnicodeIdentifierPart(lastChar)) {
                    break;
                }
            }
            // Remove trailing delimiter only if it is not block end
            boolean isBlockQuery = false;
            String[][] ORACLE_BEGIN_END_BLOCK = new String[0][];
            getBlockBoundStrings();
            String trimmed = sql.substring(0, sql.length() - statementDelimiter.length());
            {
                String test = trimmed.toUpperCase().trim();
                String[][] blockBoundStrings = ORACLE_BEGIN_END_BLOCK;
                if (blockBoundStrings != null) {
                    for (String[] blocks : blockBoundStrings) {
                        int endIndex = test.lastIndexOf(blocks[1]);
                        if (endIndex > 0) {
                            // This is a block query if it ends with 'END' or with 'END id'
                            if (test.endsWith(blocks[1])) {
                                isBlockQuery = true;
                                break;
                            } else {
                                String afterEnd = test.substring(endIndex + blocks[1].length()).trim();
                                if (afterEnd.chars().noneMatch(Character::isWhitespace)) {
                                    isBlockQuery = true;
                                    break;
                                }
                            }
                        }
                    }
                }
            }
            if (!isBlockQuery) {
                sql = trimmed;
            }
            break;
        }
        return sql + trailingSpaces;
    }

    @Override
    public String[] getScriptDelimiters() {
        return SQLConstants.DEFAULT_SCRIPT_DELIMITER; //$NON-NLS-1$
    }

    @Override
    public String[][] getBlockBoundStrings() {
        return new String[][]{
                {SQLConstants.BLOCK_BEGIN, SQLConstants.BLOCK_END},
                {"IF", SQLConstants.BLOCK_END},
                {"LOOP", SQLConstants.BLOCK_END + " LOOP"},
                {SQLConstants.KEYWORD_CASE, SQLConstants.BLOCK_END + " " + SQLConstants.KEYWORD_CASE},
        };
    }
}
